package com.heima.article.listener;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.heima.article.stream.KafkaStreamListener;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleBehaviorAggsData;
import com.heima.model.mess.ArticleBehaviorData;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;

/**
 * 聚合统计文章行为梳理的流程计算程序
 */
@Slf4j
@Component
public class ApArticleHotStreamListener implements KafkaStreamListener<KStream<String,String>> {

    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_IN_TOPIC; //入口Topic名称
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_OUT_TOPIC; //出口Topic名称
    }

    /**
     * 流式计算：在单位时间窗口内快速聚合统计出文章行为的数量
     * @param kStream
     * @return
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> kStream) {

        KStream<String, String> map = kStream.flatMapValues(value -> {
                    //value数据是原始消息数据，本案例中格式：{"articleId":123, "type": "LIKES"}
                    log.info("[实时聚合统计文章行为数量-第1阶段]value:{}", value);

                    ArticleBehaviorData data = JSON.parseObject(value, ArticleBehaviorData.class);
                    String val = data.getArticleId() + ":" + data.getType(); //拼接的值格式：123:LIKES

                    //第1阶段的目的：将原始数据转为集合数据，数据流转到第2阶段处理，集合中每个元素都会是将来被聚合的key
                    return Arrays.asList(val);
                }).map((key, value) -> {
                    //key是空值，value是第1阶段流转过来的列表中每个元素
                    log.info("[实时聚合统计文章行为数量-第2阶段]value:{}", value);

                    //第2阶段目的：指定列表中的每一个元素当做被聚合的key
                    return new KeyValue<>(value, null);
                }).groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .count(Materialized.as("hot-article-compute" + IdUtil.fastSimpleUUID()))
                .toStream()
                .map((key, value) -> {
                    //key是包含了键的对象，value是聚合的数量
                    log.info("[实时聚合统计文章行为数量-第3阶段]key:{},value:{}", key.key(), value);

                    //key: 123:LIKES   value: 100

                    //第3阶段目的：指定对应的数据给SinkProcessor，我们希望value的值不再是聚合数量，应该是包含了聚合结果的对象的JSON字符串
                    //JSON字符串格式：{"articleId":123, "like":100, "view":200, "collect":300, "comment": 400}
                    return new KeyValue<>(key.key(), handleValue(key.key(),value));
                });

        return map;
    }

    /**
     * 基于聚合结果的key和value转换为聚合结果对象的JSON字符串：目的是希望消费者处理起来更加方便
     * @param key   值格式：123:LIKES
     * @param value 聚合的行为数量
     * @return
     */
    private String handleValue(String key, Long value) {

        //1.取出key里的articleId和行为类型
        String[] keyArr = key.split(":");
        Long articleId = Long.valueOf(keyArr[0]);//文章id
        String type = keyArr[1];//行为类型

        //2.构建聚合结果对象
        ArticleBehaviorAggsData data = new ArticleBehaviorAggsData();
        data.setArticleId(articleId);//被聚合的文章id

        //设置阅读行为聚合结果数量
        if(type.equals(ArticleBehaviorData.ArticleBehaviorType.VIEWS.name())){
            data.setView(value.intValue());
        } else if(type.equals(ArticleBehaviorData.ArticleBehaviorType.LIKES.name())){
            //设置点赞行为聚合结果数量
            data.setLike(value.intValue());
        } else if(type.equals(ArticleBehaviorData.ArticleBehaviorType.COMMENT.name())){
            //设置评论行为聚合结果数量
            data.setComment(value.intValue());
        } else {
            //设置收藏行为聚合结果数量
            data.setCollect(value.intValue());
        }

        //3.转换为JSON字符串
        return JSON.toJSONString(data);
    }
}
